#ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include <iostream>
#include <vector>
#include <string>
#include <cstring>
#include <cassert>
#include <functional>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <thread>
#include <condition_variable>
#include <signal.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL DBG

#define LOG(level, format, ...)                                                                                        \
    do                                                                                                                 \
    {                                                                                                                  \
        if (level < LOG_LEVEL)                                                                                         \
            break;                                                                                                     \
        time_t t = time(NULL);                                                                                         \
        struct tm *ltm = localtime(&t);                                                                                \
        char tmp[32] = {0};                                                                                            \
        strftime(tmp, 31, "%H:%M:%S", ltm);                                                                            \
        fprintf(stdout, "[%p,%s %s:%d] " format "\n", (void *)pthread_self(), tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

#define BUFFER_DEAFULT_SIZE 1024
class Buffer
{
public:
    Buffer()
        : _writer_idx(0), _reader_idx(0), _buffer(BUFFER_DEAFULT_SIZE)
    {
    }
    // 获取缓冲区的起始地址
    char *Begin()
    {
        return &*_buffer.begin();
    }
    // 获取当前写入起始地址, _buffer的空间起始地址，加上写偏移量
    char *GetWritePosion()
    {
        return Begin() + _writer_idx;
    }
    // 获取当前读取起始地址
    char *GetReadPosion()
    {
        return Begin() + _reader_idx;
    }
    // 获取缓冲区末尾空闲空间大小--写偏移之后的空闲空间, 总体空间大小减去写偏移
    uint64_t TailIdleSpace()
    {
        return _buffer.size() - _writer_idx;
    }
    // 获取缓冲区起始空闲空间大小--读偏移之前的空闲空间
    uint64_t HeadIdleSpace()
    {
        return _reader_idx;
    }
    // 获取可读数据大小 = 写偏移 - 读偏移
    uint64_t ReadAbleSize()
    {
        return _writer_idx - _reader_idx;
    }
    // 将写偏移向后移动
    void MoveWriteOffset(uint64_t len)
    {
        if (len == 0)
            return;
        // 向后移动的大小不能大于后沿空间大小
        assert(len <= TailIdleSpace());
        _writer_idx += len;
    }
    // 将读偏移向后移动
    void MoveReadOffset(uint64_t len)
    {
        // 读偏移量不能大于可读空间大小
        if (len == 0)
            return;
        assert(len <= ReadAbleSize());
        _reader_idx += len;
    }
    // 确保可写空间足够（整体空闲空间够了就移动数据，否则就扩容）
    void EnsureWriteSpace(uint64_t len)
    {
        // 如果写入的数据长度小与后沿空间大小直接返回
        if (len <= TailIdleSpace())
            return;
        // 如果整体空闲大小足够就移动数据
        if (TailIdleSpace() + HeadIdleSpace() >= len)
        {
            uint64_t size = ReadAbleSize();
            std::copy(GetReadPosion(), GetReadPosion() + size, Begin());
            _reader_idx = 0;    // 将读偏移归0
            _writer_idx = size; // 将写位置置为可读数据大小， 因为当前的可读数据大小就是写偏移量
            return;
        }
        // 整体空闲空间不够就扩容
        _buffer.resize(len - TailIdleSpace());
    }
    // 写入数据
    void Write(const void *buf, uint64_t len)
    {
        if (len <= 0)
            return;
        EnsureWriteSpace(len);
        std::copy((char *)buf, (char *)buf + len, GetWritePosion());
    }
    // 写入数据并将写入位置向后移动
    void WriteAndPush(const void *buf, uint64_t len)
    {
        Write(buf, len);
        MoveWriteOffset(len);
    }
    // 写入字符串
    void WriteString(const std::string &str)
    {
        Write(str.c_str(), str.size());
    }
    // 写入字符串并且移动
    void WriteStringAndPush(const std::string &str)
    {
        WriteString(str);
        MoveWriteOffset(str.size());
    }
    // 写入一个Buffer对象的数据
    void WriteBuffer(Buffer &buff)
    {
        Write(buff.GetReadPosion(), buff.ReadAbleSize());
    }
    // 写入一个Buffer对象的数据并移动
    void WriteBufferAndPush(Buffer &buff)
    {
        WriteBuffer(buff);
        MoveWriteOffset(buff.ReadAbleSize());
    }
    // 读取数据
    void Read(void *buf, uint64_t len)
    {
        if (len == 0)
            return;
        assert(len <= ReadAbleSize());
        std::copy(GetReadPosion(), GetReadPosion() + len, (char *)buf);
    }
    // 读取数据并将读取位置向后移动
    void ReadAndPop(void *buf, uint64_t len)
    {
        Read(buf, len);
        MoveReadOffset(len);
    }
    // 将读取的数据作为字符串
    std::string ReadAsString(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }
    // 将读取的数据作为字符串并将读取位置向后移动
    std::string ReadAsStringAndPop(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }
    // 获取一行内容
    char *FindCRLF()
    {
        char *res = (char *)memchr(GetReadPosion(), '\n', ReadAbleSize());
        return res;
    }
    std::string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == nullptr)
        {
            return std::string();
        }
        return ReadAsString(pos - GetReadPosion() + 1);
    }

    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }
    // 清空缓冲区
    void Clear()
    {
        _reader_idx = 0;
        _writer_idx = 0;
    }

private:
    std::vector<char> _buffer;
    uint64_t _writer_idx;
    uint64_t _reader_idx;
};

#define DefaultBacklog 1024
class Socket
{
public:
    Socket() {}
    Socket(int fd) : _fd(fd) {}
    ~Socket()
    {
        Close();
    }

    int Fd()
    {
        return _fd;
    }
    bool Create()
    {
        _fd = ::socket(AF_INET, SOCK_STREAM, 0);
        if (_fd < 0)
        {
            ERR_LOG("socket create false!\n");
            return false;
        }
        DBG_LOG("socket create success!\n");
        return true;
    }

    bool Bind(uint16_t port, const std::string &ip = "0.0.0.0")
    {
        struct sockaddr_in local;
        local.sin_family = AF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(local);
        int ret = ::bind(_fd, (struct sockaddr *)&local, len);
        if (ret < 0)
        {
            ERR_LOG("bind false!\n");
            return false;
        }
        DBG_LOG("socket bind success!\n");
        return true;
    }

    bool Listen()
    {
        int ret = ::listen(_fd, DefaultBacklog);
        if (ret < 0)
        {
            ERR_LOG("listen false!\n");
            return false;
        }
        DBG_LOG("socket listen success!\n");
        return true;
    }

    int Accept()
    {
        struct sockaddr_in client;
        socklen_t len = sizeof(client);
        int newfd = ::accept(_fd, (struct sockaddr *)&client, &len);
        if (newfd < 0)
            return -1;
        DBG_LOG("get a new link!\n");
        return newfd;
    }

    bool Connect(uint16_t port, const std::string &ip)
    {
        struct sockaddr_in server;
        server.sin_family = AF_INET;
        server.sin_port = htons(port);
        server.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(server);
        int ret = ::connect(_fd, (struct sockaddr *)&server, len);
        if (ret < 0)
        {
            ERR_LOG("connect false!\n");
            return false;
        }
        DBG_LOG("socket connect success!\n");
        return true;
    }

    ssize_t Recv(void *buff, uint64_t len, int flag = 0)
    {
        ssize_t n = ::recv(_fd, buff, len, flag);
        if (n <= 0)
        {
            // EAGAIN 当前socket的接收缓冲区中没有数据了，在非阻塞的情况下才会有这个错误
            // EINTR  表示当前socket的阻塞等待，被信号打断了，
            if (errno == EINTR || errno == EAGAIN)
            {
                return 0;
            }
            ERR_LOG("SOCKET RECV FAILED!errno:%d\n", errno);
            return -1;
        }
        return n;
    }

    ssize_t NonBlockRecv(void *buf, size_t len)
    {
        return Recv(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前接收为非阻塞。
    }

    ssize_t Send(const void *buff, uint64_t len, int flag = 0)
    {
        ssize_t n = ::send(_fd, buff, len, flag);
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return 0;
            }
            ERR_LOG("SOCKET Send FAILED!errno:%d\n", errno);
            return -1;
        }
        return n;
    }

    ssize_t NonBlockSend(void *buff, uint64_t len)
    {
        if (len == 0)
            return 0;
        return Send(buff, len, MSG_DONTWAIT);
    }

    void SetNonBlock()
    {
        int flag = fcntl(_fd, F_GETFL, 0);
        fcntl(_fd, F_SETFL, flag | O_NONBLOCK);
    }

    void ReuseAddr()
    {
        int reuse = 1; // 启用套接字重用
        setsockopt(_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
        reuse = 1;
        setsockopt(_fd, SOL_SOCKET, SO_REUSEPORT, &reuse, sizeof(reuse));
    }

    bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = false)
    {
        if (!Create())
            return false;
        if (block_flag)
            SetNonBlock();
        if (!Bind(port, ip))
            return false;
        if (!Listen())
            return false;
        ReuseAddr();
        return true;
    }
    bool CreateClient(uint16_t port, const std::string &ip)
    {
        if (!Create())
            return false;
        if (!Connect(port, ip))
            return false;
        return true;
    }
    void Close()
    {
        if (_fd >= 0)
        {
            close(_fd);
            _fd = -1;
        }
    }

private:
    int _fd;
};

class Poller;
class EventLoop;
using EventCallBack = std::function<void()>;
class Channel
{
public:
    Channel(EventLoop *loop, int fd)
        : _fd(fd), _events(0), _revents(0), _loop(loop)
    {
    }
    // 设置事件的回调函数
    void SetReadCallBack(const EventCallBack &cb)
    {
        _read_callback = cb;
    }
    void SetWriteCallBack(const EventCallBack &cb)
    {
        _write_callback = cb;
    }
    void SetErrorCallBack(const EventCallBack &cb)
    {
        _error_callback = cb;
    }
    void SetCloseCallBack(const EventCallBack &cb)
    {
        _close_callback = cb;
    }
    void SetEventCallBack(const EventCallBack &cb)
    {
        _event_callback = cb;
    }

    int Fd()
    {
        return _fd;
    }

    uint32_t Events()
    {
        return _events;
    }

    void SetRevents(uint32_t revents)
    {
        _revents = revents;
    }
    // 当前是否监控了可读事件
    bool ReadAble()
    {
        return (_events & EPOLLIN);
    }
    // 当前是否监控了可写事件
    bool WriteAble()
    {
        return (_events & EPOLLOUT);
    }
    // 开启监控可读事件
    void EnableRead()
    {
        _events |= EPOLLIN;
        Update();
    }
    // 开启监控可写事件
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Update();
    }
    // 关闭监控可写事件
    void DisableRead()
    {
        _events &= ~EPOLLIN;
        Update();
    }
    // 关闭监控可写事件
    void DisableWrite()
    {
        _events &= ~EPOLLOUT;
        Update();
    }
    // 移除事件(从红黑树结构中移除)
    void Remove();
    void Update();
    // 关闭所有事件监控
    void DisableAll()
    {
        _events = 0;
        Update();
    }
    // 事件处理，⼀旦连接触发了事件，就调⽤这个函数，⾃⼰触发了什么事件如何处理⾃⼰决定
    void HanlderEvents()
    {
        if (_revents & EPOLLIN || _revents & EPOLLRDHUP || _revents & EPOLLPRI)
        {
            if (_read_callback)
                _read_callback();
        }
        // 可能造成连接断开的只执行一次，大不了下一次在执行
        if (_revents & EPOLLOUT)
        {
            if (_write_callback)
                _write_callback();
        }
        else if (_revents & EPOLLERR)
        {
            if (_error_callback)
                _error_callback();
        }
        else if (_revents & EPOLLHUP)
        {
            if (_close_callback)
                _close_callback();
        }
        // 刷新活跃度
        if (_event_callback)
            _event_callback();
    }

private:
    int _fd;
    EventLoop *_loop;
    uint32_t _events;  // 被监控的事件
    uint32_t _revents; // 被触发的事件

    EventCallBack _read_callback;  // 可读事件被触发的回调
    EventCallBack _write_callback; // 可写事件被触发的回调
    EventCallBack _error_callback; // 错误事件被触发的回调
    EventCallBack _event_callback; // 任意事件被触发的回调
    EventCallBack _close_callback; // 连接断开事件被触发的回调
};

#define GSIZE 1024
#define MAX_EPOLLEVENTS 1204
class Poller
{
private:
    // 对epoll的直接操作
    void Update(Channel *channel, int op)
    {
        int fd = channel->Fd();
        struct epoll_event event;
        event.data.fd = fd;
        event.events = channel->Events();
        int n = epoll_ctl(_epfd, op, fd, &event);
        if (n < 0)
        {
            // ERR_LOG("epoll_ctl false!\n");
            perror("epoll_ctl failed");
            abort(); // 退出程序
        }
    }
    // 判断channel是否添加到了事件监控中
    bool HasChannel(Channel *channel)
    {
        int fd = channel->Fd();
        auto pos = _channels.find(fd);
        if (pos == _channels.end())
            return false;
        return true;
    }

public:
    Poller()
    {
        _epfd = epoll_create(GSIZE);
        if (_epfd < 0)
        {
            ERR_LOG("epoll create false!\n");
            abort(); // 退出程序
        }
    }

    void UpdateEvents(Channel *channel)
    {
        bool ret = HasChannel(channel);
        // 不存在就添加，存在就修改
        if (!ret)
        {
            _channels.insert(std::make_pair(channel->Fd(), channel));
            Update(channel, EPOLL_CTL_ADD);
        }
        else
        {
            Update(channel, EPOLL_CTL_MOD);
        }
    }
    // 移除监控
    void RemoveEvent(Channel *channel)
    {
        // 在哈希表中删除
        int fd = channel->Fd();
        auto it = _channels.find(fd);
        if (it != _channels.end())
        {
            _channels.erase(it);
        }
        // 删除epoll对channel的关心
        Update(channel, EPOLL_CTL_DEL);
    }
    // 开始监控
    void Poll(std::vector<Channel *> *active)
    {
        int n = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);
        if (n < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ERR_LOG("epoll_wait false!\n");
            abort();
        }
        for (int i = 0; i < n; i++)
        {
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());          // 保证就绪的fd保存在poller中，否则就出问题了
            it->second->SetRevents(_evs[i].events); // 设置就绪事件
            active->push_back(it->second);
        }
    }

private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS];
    std::unordered_map<int, Channel *> _channels;
};

using FuncTask = std::function<void()>;
using ReleaseTask = std::function<void()>;
class TimerTask
{
public:
    TimerTask(uint64_t fd, uint32_t timeout, FuncTask func)
        : _fd(fd), _timeout(timeout), _func(func), _canceled(false)
    {
    }
    ~TimerTask()
    {
        if (!_canceled)
        {
            _func();
        }
        _release();
    }
    void SetRelease(const ReleaseTask &release)
    {
        _release = release;
    }
    uint32_t GetTimeout()
    {
        return _timeout;
    }
    void SetCannel()
    {
        _canceled = true;
    }

private:
    uint64_t _fd;         // 定时器标识fd
    uint32_t _timeout;    // 超时时间
    bool _canceled;       // fasle表示不取消任务，true表示取消任务
    FuncTask _func;       // 定时器过期执行的任务
    ReleaseTask _release; // 删除TimerWheel中的定时器信息
};

class TimerWheel
{
public:
    using WeakTask = std::weak_ptr<TimerTask>;
    using PtrTask = std::shared_ptr<TimerTask>;

private:
    void RemoveTask(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it != _timers.end())
        {
            _timers.erase(it);
        }
    }
    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ERR_LOG("timerfd create false!\n");
            abort();
        }
        struct itimerspec spc;
        spc.it_value.tv_sec = 1; // 第一次超时时间
        spc.it_value.tv_nsec = 0;
        spc.it_interval.tv_sec = 1; // 第一次超时以后的超时时间
        spc.it_interval.tv_nsec = 0;
        timerfd_settime(timerfd, 0, &spc, nullptr);
        return timerfd;
    }
    int ReadTimerfd()
    {
        uint64_t times = 0; // 必须是8字节
        int n = read(_timerfd, &times, 8);
        if (n < 0)
        {
            ERR_LOG("timerfd read error!\n");
            abort();
        }
        return times; // 返回超时的次数
    }

public:
    TimerWheel(EventLoop *loop)
        : _tick(0), _capacity(60), _wheel(_capacity), _timerfd(CreateTimerfd()), _loop(loop), _timer_channel(new Channel(loop, _timerfd))
    {
        _timer_channel->SetReadCallBack(std::bind(&TimerWheel::Ontime, this));
        _timer_channel->EnableRead(); // 开启对定时器文件描述符的读关心
    }
    // 定时器读事件触发执行的回调函数   1.读取timerfd清空 2.执行时间轮的对应函数
    void Ontime()
    {
        int times = ReadTimerfd();
        for (int i = 0; i < times; i++)
        {
            RunTimerTask();
        }
    }
    // 对定时器的操作大多都是对连接的操作，可能会发生多线程执行的情况，需要考虑线程安全问题
    // 这里让定时器的操作与对应的EventLoop绑定在一个线程
    void AddTimerTaskInLoop(uint64_t id, uint32_t timeout, const FuncTask func)
    {
        PtrTask pt(new TimerTask(id, timeout, func));
        pt->SetRelease(std::bind(&TimerWheel::RemoveTask, this, id));
        int pos = (_tick + timeout) % _capacity;
        _wheel[pos].push_back(pt);
        WeakTask wt = pt;
        _timers[id] = wt;
    }

    void RefreshTaskInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return; // 定时器任务不存在，刷新不了
        }
        PtrTask pt = it->second.lock(); // lock是weak_ptr second的shared_ptr
        int pos = (_tick + pt->GetTimeout()) % _capacity;
        _wheel[pos].push_back(pt);
    }

    void CannelTaskInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return;
        }
        PtrTask pt = it->second.lock();
        if (pt)
            pt->SetCannel();
    }

    void RunTimerTask()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear(); // 清空指定位置的数组，就会把数组中保存的所有管理定时器对象的shared_ptr释放掉
    }

    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it != _timers.end())
        {
            return true;
        }
        return false;
    }

    // 函数声明 定义在后面
    void AddTimerTask(uint64_t id, uint32_t timeout, FuncTask func);
    void RefreshTask(uint64_t id);
    void CannelTask(uint64_t id);

private:
    int _tick;                                      // 模拟秒针，走到那里执行哪里的任务
    int _capacity;                                  // 最大延迟时间
    std::vector<std::vector<PtrTask>> _wheel;       // 时间轮
    std::unordered_map<uint64_t, WeakTask> _timers; // 保存定时器信息

    int _timerfd;                            // 定时器文件描述符
    EventLoop *_loop;                        // 监控事件是EventLoop的子模块，需要一个eventloop
    std::unique_ptr<Channel> _timer_channel; // 管理定时器文件描述符的channel
};
class EventLoop
{
    using Functor = std::function<void()>;

private:
    static int CreateEventFd()
    {
        int fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (fd < 0)
        {
            ERR_LOG("eventfd create false!\n");
            abort();
        }
        return fd;
    }
    void RunAllTasks()
    {
        std::vector<Functor> tasks;
        {
            // 加锁保护任务迟的线程安全
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(tasks);
        }
        for (auto &task : tasks)
        {
            task();
        }
    }

    void ReadEventfd()
    {
        uint64_t res = 0;
        int ret = read(_event_fd, &res, sizeof(res));
        if (ret < 0)
        {
            // EINTR -- 被信号打断；   EAGAIN -- 表示无数据可读
            if (errno == EINTR || errno == EAGAIN)
            {
                return;
            }
            ERR_LOG("READ EVENTFD FAILED!");
            abort();
        }
        return;
    }

    void WeakUpEventFd()
    {
        uint64_t val = 1;
        int ret = write(_event_fd, &val, sizeof(val));
        if (ret < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ERR_LOG("READ EVENTFD FAILED!");
            abort();
        }
        return;
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()), _event_fd(CreateEventFd()), _channel(new Channel(this, _event_fd)), _wheel(this)
    {
        // 为_event_fd设置读回调函数，并开启读关心
        _channel->SetReadCallBack(std::bind(&EventLoop::ReadEventfd, this));
        _channel->EnableRead();
    }

    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }

    void AssertInLoop()
    {
        assert(_thread_id == std::this_thread::get_id());
    }

    void RunInLoop(const Functor &cb)
    {
        // 如果在当EventLoop对应的线程直接执行，否则加入到任务池中
        if (IsInLoop())
        {
            cb();
            return;
        }
        QueueInLoop(cb);
    }

    void QueueInLoop(const Functor &cb)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }
        WeakUpEventFd();
    }

    // 1.监控事件 2，执行事件 3.执行任务
    void Start()
    {
        while (1)
        {
            std::vector<Channel *> actives;
            _poller.Poll(&actives);
            for (const auto &e : actives)
            {
                e->HanlderEvents();
            }
            RunAllTasks();
        }
    }

    void RemoveEvent(Channel *channel)
    {
        _poller.RemoveEvent(channel);
    }

    void UpdateEvents(Channel *channel)
    {
        _poller.UpdateEvents(channel);
    }

    void AddTimerTask(uint64_t id, uint32_t timeout, const FuncTask func)
    {
        _wheel.AddTimerTask(id, timeout, func);
    }

    void RefreshTask(uint64_t id)
    {
        _wheel.RefreshTask(id);
    }

    void CannelTask(uint64_t id)
    {
        _wheel.CannelTask(id);
    }

    bool HasTimers(uint64_t id)
    {
        return _wheel.HasTimer(id);
    }

private:
    std::thread::id _thread_id; // 线程ID
    int _event_fd;
    Poller _poller;
    std::vector<Functor> _tasks; // 任务池
    std::mutex _mutex;           // 保护任务池的锁
    TimerWheel _wheel;
    std::unique_ptr<Channel> _channel; // 维护_event_fd
};
class LoopThread
{
private:
    /*用于实现_loop获取的同步关系，避免线程创建了，但是_loop还没有实例化之前去获取_loop*/
    std::mutex _mutex;             // 互斥锁
    std::condition_variable _cond; // 条件变量
    EventLoop *_loop;              // EventLoop指针变量，这个对象需要在线程内实例化
    std::thread _thread;           // EventLoop对应的线程
private:
    /*实例化 EventLoop 对象，唤醒_cond上有可能阻塞的线程，并且开始运行EventLoop模块的功能*/
    void ThreadEntry()
    {
        EventLoop loop; // 一个线程对应一个EventLoop

        {
            std::unique_lock<std::mutex> lock(_mutex); // 加锁
            _loop = &loop;
            _cond.notify_all();
        }
        loop.Start();
    }

public:
    /*创建线程，设定线程入口函数*/
    LoopThread() : _loop(NULL), _thread(std::thread(&LoopThread::ThreadEntry, this)) {}
    /*返回当前线程关联的EventLoop对象指针*/
    EventLoop *GetLoop()
    {
        EventLoop *loop = NULL;
        {
            std::unique_lock<std::mutex> lock(_mutex); // 加锁
            _cond.wait(lock, [&]()
                       { return _loop != NULL; }); // loop为NULL就一直阻塞
            loop = _loop;
        }
        return loop;
    }
};
class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop *baseloop)
        : _thread_counts(0), _next_loop_idx(0), _baseloop(baseloop)
    {
    }
    void Create()
    {
        if (_thread_counts > 0)
        {
            _threads.resize(_thread_counts);
            _loops.resize(_thread_counts);
            for (int i = 0; i < _thread_counts; i++)
            {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop();
            }
        }
    }
    void SetThreadPoolCounts(int count)
    {
        _thread_counts = count;
    }
    EventLoop *NextLoop()
    {
        if (_thread_counts == 0)
            return _baseloop;
        _next_loop_idx = (_next_loop_idx + 1) % _thread_counts;
        return _loops[_next_loop_idx];
    }

private:
    int _thread_counts;
    int _next_loop_idx;
    std::vector<LoopThread *> _threads;
    std::vector<EventLoop *> _loops;
    EventLoop *_baseloop;
};

class Any
{
private:
    class holder
    {
    public:
        virtual ~holder() {}
        virtual const std::type_info &type() = 0;
        virtual holder *clone() = 0;
    };
    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val) : _val(val) {}
        // 获取子类对象保存的数据类型
        virtual const std::type_info &type() { return typeid(T); }
        // 针对当前的对象自身，克隆出一个新的子类对象
        virtual holder *clone() { return new placeholder(_val); }

    public:
        T _val;
    };
    holder *_content;

public:
    // 构造
    Any() : _content(NULL) {};
    template <class T>
    Any(const T &val) : _content(new placeholder<T>(val)) {}
    // 拷贝构造
    Any(const Any &other)
        : _content(other._content ? other._content->clone() : NULL)
    {
        // 如果otehr不为空，那就可以拷贝，注意不能是浅拷贝，所以才设计了一个clone函数
    }
    // 析构
    ~Any()
    {
        if (_content)
            delete _content;
    }
    Any &swap(Any &other)
    {
        std::swap(_content, other._content);
        return *this;
    }
    // 赋值运算符重载
    template <class T>
    Any &operator=(const T &val)
    {
        Any(val).swap(*this);
        return *this;
    };
    Any &operator=(const Any &other)
    {
        Any(other).swap(*this);
        return *this;
    };

    // 获取子类对象中数据的指针
    template <class T>
    T *get()
    {
        // 想要获取的数据类型，必须和保存的数据类型一致
        assert(typeid(T) == _content->type());
        return &((placeholder<T> *)_content)->_val;
    }
};

class Connection;
// DISCONECTED -- 连接关闭状态；                             CONNECTING -- 连接建立成功-待处理状态
// CONNECTED -- 连接建立完成，各种设置已完成，可以通信的状态；  DISCONNECTING -- 待关闭状态
typedef enum
{
    DISCONNECTING,
    DISCONNECTED,
    CONNECTING,
    CONNECTED
} ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    uint64_t _conn_id; // 表示连接的唯一性
    // uint64_t _timer_id;  //这里为方便让定时器id与conn_id一致
    int _sockfd;
    Socket _socket;
    EventLoop *_loop;
    Channel _channel;
    ConnStatu _status;
    Buffer _in_buffer;
    Buffer _out_buffer;
    bool _enable_inactive_release;
    Any _context;
    // 如下四个回调函数是服务器模块设置的，也就是服务器组件使用者设置的
    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;

    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;

    // 删除服务器对连接的管理
    ClosedCallback _server_closed_callback;

private:
    /*channel的五个回调函数*/
    // 描述符触发可读事件后调用的函数 1.将从socket读取的数据放到输入缓冲区中 2.调用_message_callback回调
    void HanderRead()
    {
        char buf[65536];
        ssize_t ret = _socket.NonBlockRecv(buf, 65535);
        // 这里的ret等于0表示的是没有读取到数据，而并不是连接断开了，连接断开返回的是-1
        if (ret < 0)
        {
            // error读取错误
            // 不能直接断开连接,还要看接收缓冲区是否还有数据需要处理
            return ShutDownInLoop();
        }
        // 将数据放入输入缓冲区,写入之后顺便将写偏移向后移动
        _in_buffer.WriteAndPush(buf, ret);
        // 调用_message_callback回调处理业务
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
    }
    // 描述符可写事件触发后调用的函数，将发送缓冲区中的数据进行发送
    void HanderWrite()
    {
        // 将发送缓冲区的数据发送
        ssize_t ret = _socket.NonBlockSend(_out_buffer.GetReadPosion(), _out_buffer.ReadAbleSize());
        if (ret < 0)
        {
            // 发送错误就该关闭连接了，
            if (_in_buffer.ReadAbleSize() > 0)
            {
                if (_message_callback)
                    _message_callback(shared_from_this(), &_in_buffer);
            }
            return Release();
        }
        // 移动发送缓冲区的读开始位置
        _out_buffer.MoveReadOffset(ret);
        if (_out_buffer.ReadAbleSize() == 0)
        {
            // 没有数据需要发送了，关闭写事件关心
            _channel.DisableWrite();
            // 如果输出缓冲区没有数据并且状态为待关闭，那就直接关闭连接
            if (_status == DISCONNECTING)
                return Release();
        }
    }

    void HanderClose()
    {
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        return Release();
    }

    void HanderError()
    {
        return HanderClose();
    }

    // 如果启动非活跃自动删除连接 1.刷新活跃度 2.执行用户设置的event_callback
    void HanderEvent()
    {
        if (_enable_inactive_release == true)
            _loop->RefreshTask(_conn_id);

        if (_event_callback)
        {
            _event_callback(shared_from_this());
        }
    }

    // 连接获取之后，所处的状态下要进行各种设置（启动读监控,调用回调函数）
    void EstablishedInLoop()
    {
        // 1. 修改连接状态；  2. 启动读事件监控；  3. 调用回调函数
        assert(_status == CONNECTING); // 当前的状态必须一定是上层的半连接状态
        _status = CONNECTED;           // 当前函数执行完毕，则连接进入已完成连接状态
        _channel.EnableRead();

        if (_connected_callback)
            _connected_callback(shared_from_this());
    }

    void ShutDownInLoop()
    {
        _status = DISCONNECTING; // 将状态设置为待关闭
        // 如果输入缓冲区还有数据就处理一下
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        // 要么是处理写的时候出错
        if (_out_buffer.ReadAbleSize() > 0)
        {
            if (_channel.WriteAble() == false)
                _channel.EnableWrite();
        }
        if (_out_buffer.ReadAbleSize() == 0)
        {
            Release();
        }
    }

    void ReleaseInLoop()
    {
        // 1.修改状态
        _status = DISCONNECTED;
        // 2.关闭对事件的监控
        _channel.Remove();
        // 3.关闭文件描述符
        _socket.Close();
        // 4.如果定时器任务存在，则关闭
        if (_loop->HasTimers(_conn_id))
            CancelInactiveReleaseInLoop();
        // 5.回调函数
        if (_closed_callback)
            _closed_callback(shared_from_this());
        if (_server_closed_callback)
            _server_closed_callback(shared_from_this());
    }

    // 这个接口并不是实际的发送接口，而只是把数据放到了发送缓冲区，启动了可写事件监控
    void SendInLoop(Buffer &buf)
    {
        if (_status == DISCONNECTED)
            return;
        _out_buffer.WriteBufferAndPush(buf);
        if (_channel.WriteAble() == false)
        {
            _channel.EnableWrite();
        }
    }

    // 启动非活跃连接超时释放规则
    void EnableInactiveReleaseInLoop(int sec)
    {
        // 1. 将判断标志 _enable_inactive_release 置为true
        _enable_inactive_release = true;
        // 2. 如果当前定时销毁任务已经存在，那就刷新延迟一下即可
        if (_loop->HasTimers(_conn_id))
        {
            return _loop->RefreshTask(_conn_id);
        }
        // 3. 如果不存在定时销毁任务，则新增
        _loop->AddTimerTask(_conn_id, sec, std::bind(&Connection::Release, this));
    }

    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if (_loop->HasTimers(_conn_id))
        {
            _loop->CannelTask(_conn_id);
        }
    }

    void UpgradeInLoop(const Any &context,
                       const ConnectedCallback &conn,
                       const MessageCallback &msg,
                       const ClosedCallback &closed,
                       const AnyEventCallback &event)
    {
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }

public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
        : _conn_id(conn_id), _sockfd(sockfd), _socket(sockfd), _loop(loop), _channel(_loop, sockfd), _enable_inactive_release(false), _status(CONNECTING)
    {
        _channel.SetReadCallBack(std::bind(&Connection::HanderRead, this));
        _channel.SetWriteCallBack(std::bind(&Connection::HanderWrite, this));
        _channel.SetCloseCallBack(std::bind(&Connection::HanderClose, this));
        _channel.SetErrorCallBack(std::bind(&Connection::HanderError, this));
        _channel.SetEventCallBack(std::bind(&Connection::HanderEvent, this));
    }
    ~Connection() { DBG_LOG("RELEASE CONNECTION:%p", this); }
    // 获取管理的文件描述符
    int Fd() { return _sockfd; }
    // 获取连接ID
    int Id() { return _conn_id; }
    // 是否处于CONNECTED状态
    bool Connected() { return (_status == CONNECTED); }
    // 设置上下文--连接建立完成时进行调用
    void SetContext(const Any &context) { _context = context; }
    // 获取上下文，返回的是指针
    Any *GetContext() { return &_context; }
    void SetConnectedCallback(const ConnectedCallback &connected_callback)
    {
        _connected_callback = connected_callback;
    }
    void SetMessageCallback(const MessageCallback &message_callback)
    {
        _message_callback = message_callback;
    }
    void SetClosedCallback(const ClosedCallback &closed_callback)
    {
        _closed_callback = closed_callback;
    }
    void SetAnyEventCallback(const AnyEventCallback &event_callback)
    {
        _event_callback = event_callback;
    }

    void SetSrvClosedCallback(const ClosedCallback &server_closed_callback)
    {
        _server_closed_callback = server_closed_callback;
    }

    // 连接建立就绪后，进行channel回调设置，启动读监控，调用_connected_callback
    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }
    // 发送数据，将数据放到发送缓冲区，启动写事件监控
    void Send(const char *data, size_t len)
    {
        // 外界传入的data，可能是个临时的空间，我们现在只是把发送操作压入了任务池，有可能并没有被立即执行
        // 因此有可能执行的时候，data指向的空间有可能已经被释放了。
        Buffer buf;
        buf.WriteAndPush(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }

    // 提供给组件使用者的关闭接口--并不实际关闭，需要判断有没有数据待处理
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }
    void Release()
    {
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }
    // 启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
    void EnableInactiveRelease(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    // 取消非活跃销毁
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    // 切换协议---重置上下文以及阶段性回调处理函数 -- 而是这个接口必须在EventLoop线程中立即执行
    // 防备新的事件触发后，处理的时候，切换任务还没有被执行--会导致数据使用原协议处理了。
    void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg,
                 const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
    }
};

class Acceptor
{
    using AcceptorCallBack = std::function<void(int)>;

private:
    int CreateServer(uint32_t port)
    {
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.Fd();
    }

public:
    Acceptor(EventLoop *loop, int port)
        : _loop(loop), _socket(CreateServer(port)), _channel(_loop, _socket.Fd())
    {
        _channel.SetReadCallBack(std::bind(&Acceptor::HandleRead, this));
        // 构造函数中不能打开对监听套接字的读关心。因为设置读事件回调在后，如果刚打开读关心就有连接来了，会导致新连接无法被处理
        // 所以一定要在设置好回调函数后在打开读事件关心
    }
    void HandleRead()
    {
        int newfd = _socket.Accept();
        if (newfd < 0)
        {
            return;
        }
        if (_acceptorcallback)
            _acceptorcallback(newfd);
    }

    void SetAcceptorCallBack(const AcceptorCallBack &cb)
    {
        _acceptorcallback = cb;
    }

    void Listen()
    {
        _channel.EnableRead();
    }

private:
    Socket _socket;
    EventLoop *_loop;
    Channel _channel;

    AcceptorCallBack _acceptorcallback;
};

class TcpServer
{
private:
    uint64_t _next_id;
    uint16_t _port;
    uint32_t _timeout;
    bool _enable_inactive_release; // 是否开启非活跃自动释放连接的标志
    EventLoop _baseloop;
    Acceptor _acceptor;
    LoopThreadPool _pool;
    std::unordered_map<uint64_t, PtrConnection> _conns;

    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    using Functor = std::function<void()>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;

private:
    void RunAfterInLoop(const Functor &task, int delay)
    {
        _next_id++;
        _baseloop.AddTimerTask(_next_id, delay, task);
    }
    void NewConnection(int fd)
    {
        _next_id++;
        PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));
        conn->SetMessageCallback(_message_callback);
        conn->SetConnectedCallback(_connected_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetAnyEventCallback(_event_callback);
        conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if (_enable_inactive_release)
            conn->EnableInactiveRelease(_timeout);
        conn->Established();
        _conns.insert(std::make_pair(_next_id, conn));
    }
    void RemoveConnectionInLoop(const PtrConnection &conn)
    {
        auto it = _conns.find(conn->Id());
        if (it != _conns.end())
        {
            _conns.erase(it);
        }
    }
    void RemoveConnection(const PtrConnection &conn)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }

public:
    TcpServer(int port)
        : _port(port), _next_id(0), _timeout(0), _enable_inactive_release(false), _acceptor(&_baseloop, port), _pool(&_baseloop)
    {
        _acceptor.SetAcceptorCallBack(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.Listen();
    }
    // 设置从属线程数量
    void SetThreadCount(int count)
    {
        return _pool.SetThreadPoolCounts(count);
    }
    // 回调函数设置
    void SetConnectedCallback(const ConnectedCallback &connected_callback)
    {
        _connected_callback = connected_callback;
    }
    void SetMessageCallback(const MessageCallback &message_callback)
    {
        _message_callback = message_callback;
    }
    void SetClosedCallback(const ClosedCallback &closed_callback)
    {
        _closed_callback = closed_callback;
    }
    void SetAnyEventCallback(const AnyEventCallback &event_callback)
    {
        _event_callback = event_callback;
    }
    void EnableInactiveRelease(uint32_t timeout)
    {
        _timeout = timeout;
        _enable_inactive_release = true;
    }
    // 用于添加一个定时任务
    void RunAfter(const Functor &task, int delay)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
    }
    void Start()
    {
        _pool.Create();
        _baseloop.Start();
    }
};

// 如下设计是为了让对定时器的操作与对应的EventLoop的线程保持一致，保护线程安全
void Channel::Remove()
{
    _loop->RemoveEvent(this);
}
void Channel::Update()
{
    _loop->UpdateEvents(this);
}

void TimerWheel::AddTimerTask(uint64_t id, uint32_t timeout, const FuncTask func)
{
    _loop->RunInLoop(std::bind(&TimerWheel::AddTimerTaskInLoop, this, id, timeout, func));
}

void TimerWheel::RefreshTask(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::RefreshTaskInLoop, this, id));
}

void TimerWheel::CannelTask(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::CannelTaskInLoop, this, id));
}

class NetWork
{
public:
    NetWork()
    {
        DBG_LOG("SIGPIPE INIT");
        signal(SIGPIPE, SIG_IGN);
    }
};
static NetWork nw;
#endif